package com.niit.streaming

import java.net.URI

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}

object Spark_stream_Close {

  def main(args: Array[String]): Unit = {
//    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("window")
//    val ssc = new StreamingContext(sparkConf,Seconds(3))
      //从检查点中恢复数据
      val ssc = StreamingContext.getActiveOrCreate("BD1",()=>{
      val sparkConf = new SparkConf().setMaster("local[*]").setAppName("window")
      val ssc = new StreamingContext(sparkConf,Seconds(3))
      val lines = ssc.socketTextStream("localhost",9999)
      val wordOne = lines.map( (_,1))
      wordOne.print()
      ssc
})


    ssc.sparkContext.setLogLevel("ERROR")
    ssc.checkpoint("BD1")



   ssc.start()

    new Thread(new Runnable {
      override def run(): Unit = {

        while (true){//目的就是为了不停获取开关状态
            //如果要执行ssc.stop，必须满足一个条件才可以关闭，那么我们把这个条件称之为 开关
            /* 开关条件必须在外部存储--》HDFS MYSQL File
               HDFS-->如果当HDFS存在开关文件，就执行ssc.stop
               MYSQL--> 当MYSQL某个数据表中某个字段值为 off时候就执行ssc.stop
             */

          Thread.sleep(2000)
          //Spark 连接 HDFS
          val fs = FileSystem.get(new URI("hdfs://node1:8020"),new Configuration(),"root")
          //判断开关文件是否存在
          val bool:Boolean =  fs.exists(new Path("hdfs://node1:8020/stopSpark.txt"))
          //获取SparkStreaming的状态。 如果开关文件存在，并且SparkStreaming为激活状态 就去关闭
          val state = ssc.getState();
          if(bool){
            if(state == StreamingContextState.ACTIVE){
              /*
                第一个参数:是否关闭SparkContext
                第二个参数：是否优雅的关闭
             */
              ssc.stop(true,true)//完成了优雅的关闭

              System.exit(0)
            }

          }


        }

      }
    })

   ssc.awaitTermination()

  }


}
